package com.yunhe.common.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.exception.code.ErrorCode;
import com.yunhe.common.util.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.net.URI;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 实时数据WebSocket处理类
 * <p>主要应用于实时值类型的长连接，如量测实时值、自定义实时值、告警实时值等
 * <p>API中的“点号”不只是量测点号，也可以是自定义的任何key
 * <p>可以通过覆盖getPointField等方法来重新定义从客户端传递过来的key名称，
 * 默认点号key为“pointNumbers”，分割符为“,”
 *
 * @author liuronglei
 */
public abstract class WebsocketRealtimeHandler extends TextWebSocketHandler {
    private static final String FIELD_POINT = "pointNumbers";
    private static final String SPLIT_STR = ",";
    private static final ObjectMapper mapper = new ObjectMapper();
    // 数据结构枚举定义
    public enum DataStructure {
        Object,
        List,
        ObjectMap,
        ListMap
    }

    /**
     * id和点号对应的map
     */
    private Map<String, List<String>> idPointsMap = new ConcurrentHashMap<>();
    /**
     * 点号和session对应的map
     */
    private volatile Map<String, List<WebSocketSession>> pointSessionsMap = new ConcurrentHashMap<>();
    /**
     * 点号和当前等待发送次数对应的Map，用于延迟发送
     */
    private volatile ConcurrentMap<String, Object> pointPile = new ConcurrentHashMap<>();
    /**
     * 点号和当前是否正在锁定中对应的Map，用于延迟发送
     */
    private volatile ConcurrentMap<String, Boolean> pointHolding = new ConcurrentHashMap<>();

    @Autowired
    private WebsocketIncrementThread websocketIncrementThread;

    /**
     * 建立连接
     * @param session 连接session
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        this.afterEstablished(session);
    }

    /**
     * 断开连接
     * @param session 连接session
     * @param status  状态
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        // 移除所有session相关内容
        clearSessionPoints(session);
        this.afterClosed(session, status);
    }

    /**
     * 接收消息
     * @param session 连接session
     * @param message 接收到的消息
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        String payload = message.getPayload();
        List<String> points = new CopyOnWriteArrayList<>();
        try {
            HashMap data = mapper.readValue(payload, HashMap.class);
            Object pointsObj = data.get(this.getPointField());
            if (pointsObj != null) {
                // 先移除session相关内容
                clearSessionPoints(session);
                // 重新添加session相关内容
                String pointsStr = (String) pointsObj;
                String[] pointStrArr = pointsStr.split(this.getSplitString());
                for (String point : pointStrArr) {
                    if (StringUtil.isEmpty(point)) {
                        continue;
                    }
                    points.add(point);
                    List<WebSocketSession> sessions = pointSessionsMap.computeIfAbsent(point, k -> new CopyOnWriteArrayList<>());
                    sessions.add(session);
                }
                idPointsMap.put(session.getId(), points);
            }
        } catch (Exception e) {
            clearSessionPoints(session);
            try {
                websocketIncrementThread.sendError(session, ErrorCode.CODE.METHOD_ARGUMENT_NOTVALID);
            } catch (Exception ignored) {

            }
        }
        this.handleMessage(session, message, points);
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param mySession 要发送消息的session，如果需要按照point自动查看哪些要发送的，传递null
     */
    public void sendMessage(Object content, WebSocketSession mySession) {
        sendMessage(content, mySession, null);
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param mySession 要发送消息的session，如果需要按照point自动查看哪些要发送的，传递null
     * @param pointList key列表
     */
    public void sendMessage(Object content, WebSocketSession mySession, List<String> pointList) {
        // 如果key为空，则组建key列表为所有已连接的key
        if (pointList == null) {
            pointList = getAllPoints();
        }
        if (refreshInterval() == 0L) {
            this.sendMessageFinal(content, mySession, pointList);
        } else {
            Map<WebSocketSession, List<String>> sessionPointMap = getSessionPointMap(mySession, pointList);
            if (sessionPointMap.size() > 0) {
                sessionPointMap.forEach((session, points) -> {
                    String key = session.getId();
                    if (pointHolding.containsKey(key) && pointHolding.get(key)) {
                        pointPile.put(key, content);
                    } else {
                        doProcess(content, session, points);
                    }
                });
            }
        }
    }

    /**
     * 执行消息发送
     * @param content 消息内容
     * @param mySession 要发送消息的session
     * @param pointList key列表
     */
    private void doProcess(Object content, WebSocketSession mySession, List<String> pointList) {
        String key = mySession.getId();
        pointHolding.put(key, true);
        this.sendMessageFinal(content, mySession, pointList);
        holdPile(content, mySession, pointList);
    }

    /**
     * 对于新收到的消息进行延迟发送
     * @param content 消息内容
     * @param mySession 要发送消息的session
     * @param pointList key列表
     */
    private void holdPile(Object content, WebSocketSession mySession, List<String> pointList) {
        String key = mySession.getId();
        Timer timer = new Timer();
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                if (pointPile.containsKey(key)) {
                    doProcess(pointPile.get(key), mySession, pointList);
                    pointPile.remove(key);
                } else {
                    pointHolding.remove(key);
                }
            }
        };
        timer.schedule(timerTask, refreshInterval());
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param mySession 要发送消息的session
     * @param pointList key列表
     */
    @SuppressWarnings("unchecked")
    private void sendMessageFinal(Object content, WebSocketSession mySession, List<String> pointList) {
        // 获得发送内容的对象类型
        DataStructure dataStructure = resultDataStructure(content);
        // 组建session和要发送实时值的映射
        Map<WebSocketSession, Object> sessionResultMap = new HashMap<>();
        List<WebSocketSession> sessions = null;
        for (String point : pointList) {
            if (mySession == null) {
                sessions = pointSessionsMap.get(point);
            } else if (sessions == null) {
                sessions = Collections.singletonList(mySession);
            }
            if (sessions != null) {
                for (WebSocketSession session : sessions) {
                    if (dataStructure == DataStructure.Object || dataStructure == DataStructure.List) {
                        sessionResultMap.put(session, content);
                    } else if (dataStructure == DataStructure.ObjectMap || dataStructure == DataStructure.ListMap) {
                        Map<Object, Object> result = (Map<Object, Object>)sessionResultMap.computeIfAbsent(session, k -> new HashMap<>());
                        result.put(point, ((Map<Object, Object>)content).get(point));
                    }
                }
            }
        }

        // 循环每个session，进行实时值发送
        for (Map.Entry<WebSocketSession, Object> entry : sessionResultMap.entrySet()) {
            try {
                WebSocketSession session = entry.getKey();
                if (!session.isOpen()) {
                    clearSessionPoints(session);
                    continue;
                }
                Object result = entry.getValue();
                beforeSend(session, result);
                if (dataStructure == DataStructure.ObjectMap) {
                    websocketIncrementThread.sendMessageObjectMap(session, (Map)result);
                } else if (dataStructure == DataStructure.ListMap) {
                    websocketIncrementThread.sendMessageListMap(session, (Map)result);
                } else if (dataStructure == DataStructure.List) {
                    websocketIncrementThread.sendMessageList(session, (List)result);
                } else {
                    websocketIncrementThread.sendMessage(session, result);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 清空session对应的点号信息
     * @param session 连接session
     */
    private void clearSessionPoints(WebSocketSession session) {
        List<String> points = idPointsMap.get(session.getId());
        if (points != null && points.size() > 0) {
            for (String point : points) {
                List<WebSocketSession> sessions = pointSessionsMap.get(point);
                if (sessions != null) {
                    sessions.remove(session);
                    // 移除session后，如果列表为空了，将key也一并移除
                    if (sessions.size() == 0) {
                        pointSessionsMap.remove(point);
                    }
                }
            }
        }
        idPointsMap.remove(session.getId());
        pointPile.remove(session.getId());
        pointHolding.remove(session.getId());
    }

    /**
     * 获得消息内容的对象类型
     * @param content 消息内容
     */
    @SuppressWarnings("unchecked")
    private DataStructure resultDataStructure(Object content) {
        DataStructure dataStructure = getDataStructure();
        if (dataStructure == null) {
            if (content instanceof List) {
                dataStructure = DataStructure.List;
            } else if (content instanceof Map) {
                dataStructure = DataStructure.ObjectMap;
                Map<Object, Object> contentMap = (Map<Object, Object>)content;
                for (Map.Entry<Object, Object> entry : contentMap.entrySet()) {
                    if (entry.getValue() instanceof List) {
                        dataStructure = DataStructure.ListMap;
                    }
                    break;
                }
            } else {
                dataStructure = DataStructure.Object;
            }
        }
        return dataStructure;
    }

    /**
     * 获得本次发送的session和对应key的Map
     * @param pointList 点号列表
     */
    private Map<WebSocketSession, List<String>> getSessionPointMap(WebSocketSession mySession, List<String> pointList) {
        Map<WebSocketSession, List<String>> result = new HashMap<>();
        if (mySession != null) {
            result.put(mySession, pointList);
        } else {
            for (String point : pointList) {
                if (pointSessionsMap.containsKey(point)) {
                    List<WebSocketSession> sessions = pointSessionsMap.get(point);
                    if (sessions != null && sessions.size() > 0) {
                        sessions.forEach(session -> result.computeIfAbsent(session, v -> new ArrayList<>()).add(point));
                    }
                }

            }
        }
        return result;
    }

    /**
     * 获得所有已连接的key列表
     */
    public List<String> getAllPoints() {
        List<String> pointList = new ArrayList<>();
        for (Map.Entry<String, List<WebSocketSession>> entry : pointSessionsMap.entrySet()) {
            String onePoint = entry.getKey();
            pointList.add(onePoint);
        }
        return pointList;
    }

    /**
     * 返回给客户端的数据结构
     * 默认根据返回结果的对象类型来定义
     * 如果需要定制，可以重写该方法
     */
    public DataStructure getDataStructure() {
        return null;
    }

    /**
     * 获得从客户端传递过来的点号key
     * 如果需要定制，可以重写该方法
     */
    public String getPointField() {
        return FIELD_POINT;
    }

    /**
     * 获得从客户端传递过来的分隔符
     * 如果需要定制，可以重写该方法
     */
    public String getSplitString() {
        return SPLIT_STR;
    }

    /**
     * 数据刷新间隔_毫秒
     * 主要是为了避免刷新过于频繁，前端压力过大，默认没有刷新间隔（即数据变化立即刷新），如果需要定义，可以重写该方法
     */
    public Long refreshInterval() {
        return 0L;
    }

    /**
     * 获得URL中的参数值
     * @param session 连接session
     * @param key 参数Key
     */
    public String getUrlParam(WebSocketSession session, String key) {
        URI uri = session.getUri();
        String query = uri.getQuery();
        if (!StringUtil.isEmpty(query)) {
            String[] arr = query.split("&");
            if (arr.length > 0) {
                Map<String, String> params = new HashMap<>();
                for (String str : arr) {
                    String[] brr = str.split("=");
                    if (brr.length > 1) {
                        params.put(brr[0], brr[1]);
                    }
                }
                return params.get(key);
            }
        }
        return null;
    }

    /**
     * 发送前的数据处理
     * 诸如中英文翻译之类的处理工作
     */
    public abstract void beforeSend(WebSocketSession session, Object content);

    /**
     * 建立连接
     * @param session 连接session
     */
    public abstract void afterEstablished(WebSocketSession session);

    /**
     * 断开连接
     * @param session 连接session
     * @param status  状态
     */
    public abstract void afterClosed(WebSocketSession session, CloseStatus status);

    /**
     * 接收消息
     * 如果在接收消息后，马上返回给客户端最新的值，则需要在实现方法中查询最新的值，并调用sendMessage()方法进行发送
     * @param session 连接session
     * @param message 接收到的消息
     */
    public abstract void handleMessage(WebSocketSession session, TextMessage message, List<String> points);

}